<?php

namespace app\data\util;
use think\facade\Config;
/**
 * 创建连接-->创建channel-->创建交换机对象-->发送消息
 */
class Rabbitmq
{
    private $_config =['host' => '', 'port' => '5672', 'vhost' => '/', 'login' => '', 'password' => ''];
    private static $_connect_static_array = [];
    private static $_channel_static_array = [];
    private static $_exchange_static_array = [];
    private static $_queue_static_array = [];
    private $_connect = null;
    private $_exchange = null;
    private $_channel = null;
    private $_queue = null;
    private $deliveryTag = 0;
    private $rabbitmq_conf = "default";
    private $queue_name = "";

    public $task_execute_time = 600;
    public $task_start_time;

    private static $_instance_static_array = [];

    function  __construct($confDefault)
    {

        $this->config = Config::get('rabbitmq.' . $confDefault);
        $this->_config      = $this->config;
        $this->_init_connect();   ##初始化连接
    }


    /**
     * 生成配置键值
     * @author xx
     * @Date 2024/3/12 14:26
     * @return string
     */
    private function _get_config_md5_key(){
        return md5(json_encode($this->_config));
    }

    /**
     * 初始化连接
     * @author xx
     * @Date 2024/3/12 14:26
     * @return string
     */
    private function _init_connect(){

        $config_key = $this->_get_config_md5_key();
        if ( !isset(self::$_connect_static_array[$config_key]) ){
            try {
                $this->_connect = new \AMQPConnection($this->_config);
                $this->_connect->connect();
                $this->_connect->setReadTimeout(0);      //AMQPConnectionException' with message '(unknown error)
                self::$_connect_static_array[$config_key] = $this->_connect;
            }catch(AMQPConnectionException $e) {
                return $e->getMessage();
            }
        }
        $this->_connect = self::$_connect_static_array[$config_key];
    }

    public static function instance($config_key) {

        if ( !isset(self::$_instance_static_array[$config_key]) ){
            self::$_instance_static_array[$config_key] = new self($config_key);
        }
        return self::$_instance_static_array[$config_key];
    }

    /**
     *  判断连接是否可用，否则重新创建新的连接
     * @author xx
     * @Date 2024/3/12 14:26
     */
    private function _check_connection()
    {
        if (empty($this->_connect) || !$this->_connect->isConnected())
        {
            $config_key = $this->_get_config_md5_key();
            unset(self::$_connect_static_array[$config_key]);
            $this->_init_connect();
        }
        return $this;
    }

    /**
     * 创建信号通道
     * @author xx
     * @Date 2024/3/12 14:26
     * @return mixed|null
     */
    public function mq_get_channel(){
        $this->_check_connection();
        $exchange_key = $this->_get_config_md5_key();
        if ( !isset(self::$_exchange_static_array[$exchange_key]) ){
            try {
                self::$_channel_static_array[$exchange_key] = new \AMQPChannel($this->_connect);
            }catch(AMQPConnectionException $e) {
                throw new \RuntimeException($e->getMessage());
            }
        }
        $this->_channel = self::$_channel_static_array[$exchange_key];
        return $this->_channel;
    }

    /**
     * 创建交换机信号
     * @author xx
     * @Date 2024/3/12 14:26
     * @param $exchange_name
     * @param $exchange_type
     * @return mixed|null
     */
    public function mq_get_exchange($exchange_name,$exchange_type){
        $exchange_key = $this->_get_config_md5_key().'_'.md5(json_encode(func_get_args()));
        if ( !isset(self::$_exchange_static_array[$exchange_key]) ){
            self::$_exchange_static_array[$exchange_key] = $this->_mq_get_exchange($exchange_name,$exchange_type);
        }
        $this->_exchange = self::$_exchange_static_array[$exchange_key];
        return $this->_exchange;
    }


    /**
     * 创建交换机
     * @author xx
     * @Date 2024/3/12 14:26
     * @param $exchange_name
     * @param $exchange_type
     * @return AMQPExchange|null
     */
    private function _mq_get_exchange($exchange_name,$exchange_type){

        try{
            $channel = $this->mq_get_channel();              //创建信号通道
            $exchange = new \AMQPExchange($channel);         //创建交换机
            $exchange->setName($exchange_name);             //交换机名称
            $exchange->setType($exchange_type);             //交换机类型-指定key
            $exchange->setFlags(AMQP_DURABLE);             //持久化
            $exchange->declareExchange();

        }catch (Exception $e){
            throw new RuntimeException($e->getMessage());
        }

        return $exchange;
    }


    /**
     * 声明队列
     * @author xx
     * @Date 2024/3/12 14:26
     * @param string $name
     * @param bool $durable
     * @param bool $priority    //设置了优先级
     * @return AMQPQueue|null
     * @throws AMQPChannelException
     * @throws AMQPQueueException
     */
    public function mq_get_queue($name="", $durable = true, $priority = false)
    {
        try {
            $name = $name;
            $this->mq_get_channel();
            $this->_queue = new \AMQPQueue($this->_channel);
            $this->_queue->setName($name);

            ////队列标识
            #AMQP_DURABLE：持久的交换和队列将在代理重新启动后幸存下来，并保留所有数据。
            #AMQP_PASSIVE：不会重新声明被动交换和队列，但是如果交换或队列不存在，则代理将引发错误。
            #AMQP_EXCLUSIVE：此标志仅对队列有效，指示只有一个客户端可以侦听并使用该队列。
            #AMQP_AUTODELETE：一旦绑定队列，交换将被删除
            # zhangji20220922,原本设置的是AMQP_DURABLE，改为AMQP_PASSIVE；原因是AMQP_DURABLE优先队列获取不到数量
            if($durable == true) $this->_queue->setFlags(AMQP_PASSIVE);
//            if($priority) {
            # x-max-priority设置该队列中的消息的优先级最大值.发布消息的时候,可以指定消息的优先级,优先级高的先被消费.如果没有设置该参数,那么该队列不支持消息优先级功能.
//                $this->_queue->declareQueue($name,true,false,false,['x-max-priority' => ['I', 9]]);
//            }else{
            $this->_queue->declareQueue();
//            }


        } catch (AMQPConnectionException $e) {
            $this->_queue = null;
            throw new Exception($e->getMessage());
        }

        return  $this->_queue;
    }

    public function mq_un_bind_queue($queueConfig){
        try {
            $queue_name = $queueConfig['queue_name'];
            $bind_key = empty($queueConfig['bind_key']) ? '' : $queueConfig['bind_key'];
            $exchange_name = empty($queueConfig['exchange_name']) ? '' : $queueConfig['exchange_name'];
            $exchange_type = empty($queueConfig['exchange_type']) ? AMQP_EX_TYPE_DIRECT : $queueConfig['exchange_type'];
            $this->mq_get_exchange($exchange_name, $exchange_type);
            $channel = $this->_channel;
            $queue = new \AMQPQueue($channel);               //创建队列
            if(isset($queueConfig['is_delay_queue'])&&$queueConfig['is_delay_queue']==1){
                $queue->setName($queue_name);                   //队列名称
                $queue->setArgument("x-message-ttl", intval($queueConfig['delay_time']));
                $queue->setArgument("x-dead-letter-exchange", $queueConfig['delay_exchange_name']);
                $queue->setArgument("x-dead-letter-routing-key",$queueConfig['delay_bind_key']);
            }else{
                $queue->setName($queue_name);                   //队列名称
            }
            $queue->setFlags(AMQP_DURABLE);                 //持久化
            $queue->declareQueue();
            $queue->unbind($exchange_name, $bind_key);        //交换机绑定key
        }catch (Exception $e){
            throw new Exception($e->getMessage());
        }
    }

    /**
     * @author xx
     * @Date 2024/3/12 16:23
     * @desc 队列绑定
     * @param $queueConfig
     * @throws AMQPChannelException
     * @throws AMQPConnectionException
     * @throws AMQPQueueException
     */
    public function mq_create_bind_queue($queueConfig){
        try {
            $queue_name = $queueConfig['queue_name'];
            $bind_key = empty($queueConfig['bind_key']) ? '' : $queueConfig['bind_key'];
            $exchange_name = empty($queueConfig['exchange_name']) ? '' : $queueConfig['exchange_name'];
            $exchange_type = empty($queueConfig['exchange_type']) ? AMQP_EX_TYPE_DIRECT : $queueConfig['exchange_type'];
            $this->mq_get_exchange($exchange_name, $exchange_type);
            $channel = $this->_channel;
            $queue = new \AMQPQueue($channel);               //创建队列
            $queue->setName($queue_name);                   //队列名称
            $queue->setFlags(AMQP_DURABLE);                 //持久化
            $queue->declareQueue();
            $queue->bind($exchange_name, $bind_key);        //交换机绑定key
        }catch (Exception $e){
            throw new Exception($e->getMessage());
        }

    }


    /**
     * 推送数据到队列上
     * @author xx
     * @Date 2024/3/12 14:26
     * @param $message
     * @param $queueConfig
     * @return bool
     * @throws Exception
     */
    public function mq_send_message($message,$queueConfig,$attributes=array(),$flags=AMQP_NOPARAM){
        $bind_key = empty($queueConfig['bind_key']) ? '' : $queueConfig['bind_key'];
        $exchange_name = empty($queueConfig['exchange_name']) ? '' : $queueConfig['exchange_name'];
        $exchange_type = empty($queueConfig['exchange_type']) ? AMQP_EX_TYPE_DIRECT : $queueConfig['exchange_type'];

        try{
            $exchange = $this->mq_get_exchange($exchange_name,$exchange_type);
            //存入的数据一定是字符串
            if (is_array($message) || is_object($message)) $message = json_encode($message);


            // 发送数据,AMQP_MANDATORY,
            // array('delivery_mode'=>2) 数据持久化   cyz-20211126 : todo 核实 目前了解之前不是持久化的队列，不能直接调整为持久化
            // priority 未指定值的消费者默认是0，数字越大优先级越高，可以使用正数和负数
            # 数据必须持久推送，如不是持久化，重启服务器后消息数据会清零
            if (!isset($attributes['delivery_mode'])) {
                $attributes['delivery_mode'] = 2;
            }
            $result = $exchange->publish($message,$bind_key,$flags,$attributes);
            if(!$result) throw new \Exception('发送MQ消息失败');
          }catch (Exception $e){
            throw new \Exception($e->getMessage());
         }
        return true;
    }




    /**
     * 消费队列
     * @author xx
     * @Date 2024/3/13 14:26
     * @param string $queueName
     * @param string $callback
     * @param bool $autoack 收到消息后,是否不需要回复确认即被认为被消费
     * @param bool $durable
     * @return mixed
     * @throws AMQPChannelException
     * @throws AMQPQueueException
     */
    public function  mq_consume_queue($queueName="", $callback = "", $autoack = true, $durable = true)
    {
        $this->mq_get_queue($queueName, $durable);

        if($autoack) {
            return $this->_queue->consume($callback, AMQP_AUTOACK);
        } else {
            return $this->_queue->consume($callback);
        }
    }

    /**
     * 获取消息队列的任务数量
     * @author xx
     * @Date 2024/3/12 14:26
     * @param string $name
     * @return int
     * @throws AMQPChannelException
     * @throws AMQPQueueException
     */
    public function mq_get_queue_num($name=""){
        try {
            $queue = $this->mq_get_queue($name);
            $messageCount = $queue->declareQueue();
        } catch (AMQPConnectionException $e) {
            return 0;
        }
        return $messageCount;
    }


    /**
     * 删除指定队列
     * @author xx
     * @Date 2024/3/12 14:26
     * @param string $name
     * @return int
     * @throws AMQPChannelException
     * @throws AMQPQueueException
     */
    public function mq_delete($name=""){
        try {
            $queue = $this->mq_get_queue($name);
            $messageCount = $queue->delete();
        } catch (AMQPConnectionException $e) {
            return 0;
        }
        return $messageCount;
    }

    /**
     * 清除指定队列内容
     * @author xx
     * @Date 2024/3/12 14:26
     * @param string $name
     * @return int
     * @throws AMQPChannelException
     * @throws AMQPQueueException
     */
    public function mq_purge($name=""){
        try {
            $queue = $this->mq_get_queue($name);
            $messageCount = $queue->purge();
        } catch (AMQPConnectionException $e) {
            return 0;
        }
        return $messageCount;
    }








    private function  get($autoack = true)
    {
        try {
            $msg = "";
            if($autoack) {
                $msg = $this->_queue->get(AMQP_AUTOACK);
            } else {
                $msg = $this->_queue->get();
            }
            if(empty($msg)) { return "";}
            $this->deliveryTag = $msg->getDeliveryTag();
            return $msg->getBody();
        } catch (AMQPConnectionException $e) {
            $this->deliveryTag = 0;
            return "";
        }
    }

    public function  ack()
    {
        try {
            $this->_queue->ack($this->deliveryTag);
        } catch (AMQPConnectionException $e) {
            $this->deliveryTag = 0;
            return false;
        }

        return true;
    }

    public function  nack()
    {
        try {
            $this->queue->nack($this->deliveryTag);
        } catch (AMQPConnectionException $e) {
            $this->deliveryTag = 0;
            return false;
        }

        return true;
    }

    /**
     * [getMsg 消费信息]
     * @param  string  $queueName [队列名称]
     * @param  boolean $autoack   [description]
     * @param  boolean $durable   [description]
     * @return [type]             [description]
     */
    public function  getMsg($queueName = '', $autoack = true, $durable = true)
    {
        $this->mq_get_queue($queueName, $durable);
        return $this->get($autoack);
    }


    public function  sendMsgFanout($msg = '', $exchangeName = '', $durable = true, $exchangeType = AMQP_EX_TYPE_FANOUT )
    {
        $this->exchange($exchangeName, $durable, $exchangeType);
        $a=$this->exchange->publish($msg,'',AMQP_MANDATORY,['delivery_mode' => 2]);
        p($a);
        die;
        // $this->connection->disconnect();
    }


    /**
     * 初始化一个守护进程
     * @throws Exception
     */
    public function forkInit()
    {
        //创建一个子进程
        $pid = pcntl_fork();

        if ($pid == -1) {
            throw new Exception('fork子进程失败');
        } elseif ($pid > 0) {
            //父进程退出,子进程变成孤儿进程被1号进程收养，进程脱离终端
            exit(0);
        }

        //创建一个新的会话，脱离终端控制，更改子进程为组长进程
        $sid = posix_setsid();
        if ($sid == -1) {
            throw new Exception('setsid fail');
        }

        //修改当前进程的工作目录，由于子进程会继承父进程的工作目录，修改工作目录以释放对父进程工作目录的占用。
        chdir('/');

        /**
         * 通过上一步，我们创建了一个新的会话组长，进程组长，且脱离了终端，但是会话组长可以申请重新打开一个终端，为了避免
         * 这种情况，我们再次创建一个子进程，并退出当前进程，这样运行的进程就不再是会话组长。
         */
        $pid = pcntl_fork();
        if ($pid == -1) {
            throw new Exception('fork子进程失败');
        } elseif ($pid > 0) {
            //再一次退出父进程，子进程成为最终的守护进程
            exit(0);
        }
        //由于守护进程用不到标准输入输出，关闭标准输入，输出，错误输出描述符
        fclose(STDIN);
        fclose(STDOUT);
        fclose(STDERR);
    }


    /**
     * @brief  getLength        获取队列的长度
     *
     * @param  mixed $queueName
     *
     * @return
     */
    public function getLength($queueName)
    {
        try{
            $channel = $this->mq_get_channel();              //创建信号通道
            $queue = new \AMQPQueue($channel);
            $queue->setFlags(AMQP_PASSIVE);
            $queue->setName($queueName);
            $count = $queue->declareQueue();
            return $count;
        }catch (Exception $e){
            throw new RuntimeException($e->getMessage());
        }

        return 0;
    }

}